//package com.study.iceberg.flink;
//
//import org.apache.flink.api.common.RuntimeExecutionMode;
//import org.apache.flink.streaming.api.datastream.DataStream;
//import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
//import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
//import org.apache.flink.table.data.GenericRowData;
//import org.apache.flink.table.data.RowData;
//import org.apache.flink.table.data.StringData;
//import org.apache.iceberg.flink.TableLoader;
//import org.apache.iceberg.flink.sink.FlinkSink;
//
//import java.util.stream.Stream;
//
//public class WriteDemo1 {
//    public static void main(String[] args) throws Exception {
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//
//        //env.setRuntimeMode(RuntimeExecutionMode.BATCH); //指定运行时模式为BATCH模式
//        // 创建待插入的测试数据DataStream
//        DataStream<RowData> inputData = env.fromElements("A", "B", "C").map(d -> {
//            // 构造函数参数指定每行数据有几列
//            GenericRowData rowData = new GenericRowData(2);
//            // 以每个element的第一个字符的ascii码作为sample表中id列的数据
//            rowData.setField(0, (int) d.charAt(0));
//            // 以每个element本身作为sample表中data列的数据
//            // flink中RowData里的字符串类型是StringData，String类型的对象不能直接强转成StringData,可以通过StringData.fromString 转换
//            rowData.setField(1, StringData.fromString(d));
//            return rowData;
//        });
//
//        /*
//        写数据还需要考虑当前用户是否对hdfs上的表目录有写权限，如果没有，学习环境直接通过hadoop fs -chmod 777 <表路径>命令修改
//         */
//        // 获取tableloader
//        TableLoader tableLoader = TableLoader.fromHadoopTable("hdfs://192.168.200.101:9820/warehouse/iceberg-hadoop/iceberg_db/sample");
//
//        FlinkSink.forRowData(inputData)
//                .tableLoader(tableLoader)
//                //        .overwrite(true) // 覆盖写，必须是batch模式
//                //.upsert(true)
//                .append();
//
//        env.execute();
//    }
//}
